#include "ThreadPoolOfCpp.h"
#include <chrono>
#include <sstream>
#include <iostream>
#include <functional>
#include <type_traits>

using namespace std;

template <class T>
void deleter(void *const obj)
{   
    // 做了一个适配，由于需要转换删除的是一个指针类型，泛型接受可以是任意类型
    // 虽然不做适配，T 如果传入的是非指针类型，在编译时会报错，以为尝试讲一个指针类型转换成一个非指针类型，并删除是非法的
    // 但是这样在使用的时候就需要注意有点麻烦，所以在这里统一加上 * 来强制使用指针，但是由于 T 可能传入指针类型、也可能是非指针类型，如：
    // int 或 int *  那么强制在指定一个指针类型如 T * 那么可能变成： int => T * => int* 是对的；int* => T* => int** 是错的了。
    // 所以需要一个适配转换，如果传入的是指针类型则移除指针只保留原本类型，如果是一个非指针类型则什么都不干。
    // remove_pointer 就可以去除掉类型里的指针类型，如果是 int => int 如果是 int * => int。所以转换玩的 Type 在来做 Type* 就能够保证一定是变量原始类型的指针了。
    typedef typename remove_pointer<T>::type Type;
    delete (Type *) obj;
    cout << "任务参数被 delete 了\n";
}

string inline getCurrentThreadId()
{
    stringstream threadId;
    threadId << this_thread::get_id();
    return threadId.str();
}

struct TData
{
    int a;
    TData(int value) : a(value) {}
    ~TData()
    {
        cout << "TData 被析构了" << endl;
    }
};

void threadPoolDemoOfCpp()
{
    cout << "线程池使用案例（基于 C++ 标准库）演示开始" << endl;
    ThreadPool threadPool(3, 10);

    cout << "---------------------------------" << endl;
    this_thread::sleep_for(chrono::seconds(5));
    for (int i = 0; i < 100; ++i)
    {
        int *data = new int(i);
        threadPool.addTask(
            [](void *const arg)
            {
                this_thread::sleep_for(chrono::seconds(1));
                int value = *(int *)arg;
                cout << "任务执行在线程ID：" + getCurrentThreadId() + "上执行，结果：" + to_string(value) + "\n";
            },
            data, deleter<decltype(data)>);

        if(i % 10 == 9)
            this_thread::sleep_for(chrono::seconds(3));
    }

    this_thread::sleep_for(chrono::seconds(15));
    
    for (int i = 0; i < 50; ++i)
    {
        TData *data = new TData(i);
        threadPool.addTask(
            [](void *const arg)
            {
                this_thread::sleep_for(chrono::seconds(1));
                TData * const data = (TData*) arg;
                data->a += 10000;
                cout << "任务执行在线程ID：" + getCurrentThreadId() + "上执行，结果：" + to_string(data->a) + "\n";
                data->a = 12;
            },
            data, deleter<decltype(data)>);
    }

    this_thread::sleep_for(chrono::seconds(20));

    cout << "线程池使用案例（基于 C++ 标准库）演示结束" << endl;
}

TaskQueue::TaskQueue()
{
}

TaskQueue::~TaskQueue()
{
    clear();
    cout << "TaskQueue 销毁了\n";
}

void TaskQueue::addTask(Task *task)
{
    queueMutex.lock();
    taskQueue.push(task);
    queueMutex.unlock();
    queueCondition.notify_one();
}

Task *TaskQueue::takeTask()
{
    lock_guard<mutex> locker(queueMutex);
    queueCondition.wait(queueMutex, [this]() -> bool
                        { return !taskQueue.empty(); });
    Task *task = taskQueue.front();
    taskQueue.pop();
    return task;
}

int TaskQueue::size()
{
    lock_guard<mutex> locker(queueMutex);
    return taskQueue.size();
}

int TaskQueue::empty()
{
    lock_guard<mutex> locker(queueMutex);
    return taskQueue.empty();
}

void TaskQueue::clear()
{
    lock_guard<mutex> locker(queueMutex);
    while (!taskQueue.empty())
    {
        Task *task = taskQueue.front();
        taskQueue.pop();
        delete task;
    }
}

ThreadPool::ThreadPool(int min, int max) : minNum(min), maxNum(max)
{
    shutdown = false;
    liveNum = 0;
    busyNum = 0;

#ifdef _WIN32
        managerThread = thread(bind(ThreadPool::managerProcess, this));
#elif __linux__ | __APPLE__
        managerThread = thread([this]()-> void { managerProcess(); });
#endif

    for (int i = 0; i < min; ++i)
    {
        createWorkerProcess();
    }
}

ThreadPool::~ThreadPool()
{
    shutdown = true;
    
    if(managerThread.joinable())
        managerThread.join();

    taskQueue.clear();
    for (int i = 0; i < liveNum; ++i)
    {
        taskQueue.addTask(new Task(false));
    }
    poolCondition.wait(
        poolMutex,
        [this]() -> bool
        {
            cout << "ThreadPool liveNum " + to_string(liveNum) + "\n";
            return liveNum <= 0;
        });

    cout << "ThreadPool 销毁了 \n";
}

void ThreadPool::addTask(TaskCallBack taskCallBack)
{
    this->addTask(taskCallBack, nullptr);
}

void ThreadPool::addTask(TaskCallBack taskCallBack, void *arg)
{
    this->addTask(taskCallBack, arg, nullptr);
}

void ThreadPool::addTask(TaskCallBack taskCallBack, void *arg, TaskDeleter deleter)
{
    taskQueue.addTask(new Task(true, taskCallBack, arg, deleter));
}

void ThreadPool::managerProcess()
{
    int NUMBER = 2;

    while (!shutdown)
    {
        this_thread::sleep_for(chrono::seconds(3));

        int queueSize = taskQueue.size();

        poolMutex.lock();
        int liveNum = this->liveNum;
        int busyNum = this->busyNum;
        poolMutex.unlock();
        
        int idleNum = liveNum - busyNum; 

        cout << "管理线程: ID=" + getCurrentThreadId() + "，开始工作:\n" 
                + "当前线程最小：" + to_string(minNum) + "\n" 
                + "当前线程最大：" + to_string(maxNum) + "\n" 
                + "当前存活线程：" + to_string(liveNum) + "\n" 
                + "当前忙碌线程：" + to_string(busyNum) + "\n" 
                + "当前闲置线程：" + to_string(idleNum) + "\n" 
                + "当前队列大小：" + to_string(queueSize) + "\n";

        if(idleNum == queueSize)
        {
            // 闲置线程数刚刚够处理任务任务队列则不做任何处理：
            // 这里可能有 2 种情况： 
            //   1. queueSize 有任务，那么闲置线程和任务数相等，刚好够处理任务，不需要新增也不能清除闲置线程
            //   2. queueSize 没有任务，那么闲置线程和任务数相等，都是 0，同样也不需要新增也不能清除闲置线程
        } 
        else if (idleNum > queueSize)
        {
            // 闲置的线程数大于任务数：处理能力已经过剩了，需要清除一些闲置线程
            //   1. 存活的线程至少保证不能低于最小线程数，准备待命。
            //   2. 如果闲置线程数量超过了存活线程的50%，也就是说闲置线程数大于忙碌线程，则释放 2 个，如果没有则不销毁一遍又有新任务加入需要重开。
            if (liveNum > minNum && liveNum > (busyNum * 2))
            {
                for (int i = 0; i < NUMBER; ++i)
                {
                    taskQueue.addTask(new Task(false));
                }
            }
        }
        else
        {
            // 闲置的线程数小于任务数：处理能力已经不足，需要增加一些线程来处理
            //   1. 增加的线程不能超过最大线程数，这里由于一次增加 2 个，所以在创建线程内还做了一次实时线程数检查以防止越界创建。
            if(liveNum < maxNum)
            {
                for (int i = 0; i < NUMBER; ++i)
                {
                    createWorkerProcess();
                }
            }
        }
    }
    cout << "管理者线程退出了\n";
}

void ThreadPool::createWorkerProcess()
{
    cout << "准备创建工作线程了\n";
    lock_guard<mutex> locker(poolMutex);
    if (liveNum >= maxNum)
        return;

#ifdef _WIN32
        thread workerThread(bind(ThreadPool::workerProcess, this));
#elif __linux__ | __APPLE__ 
        thread workerThread([this]()-> void { workerProcess(); });
#endif
    if(workerThread.joinable())
    {
        cout << "创建工作线程成功！！\n";
        workerThread.detach();
        liveNum++;
    }
}

void ThreadPool::workerProcess()
{
    cout << "工作线程，ID：" + getCurrentThreadId() + "，就绪 \n";
    while (true)
    {
        Task *task = taskQueue.takeTask();
        if (!task->needRun)
        {
            delete task;
            cout << "工作线程，ID：" + getCurrentThreadId() + "，收到了非运行任务 \n";
            // 如果线程池销毁则需要强制退出，线程池没有销毁则需要看存活的线程数是否大于最小线程数，大于则销毁，小于或等于则不销毁。
            if (!shutdown && liveNum <= minNum)
            {
                continue;
            }
            break;
        }

        cout << "工作线程，ID：" + getCurrentThreadId() + "，开始执行\n";
        poolMutex.lock();
        busyNum++;
        poolMutex.unlock();

        task->callBack(task->arg);

        poolMutex.lock();
        busyNum--;
        poolMutex.unlock();
        cout << "工作线程，ID：" + getCurrentThreadId() + "，结束执行\n";
        delete task;
    }
    
    poolMutex.lock();
    liveNum--;
    poolMutex.unlock();
    
    poolCondition.notify_one();

    cout << "工作线程，ID：" + getCurrentThreadId() + "，退出 \n";
}
